package demo.kafka.config;

import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

import java.time.LocalDateTime;
import java.util.function.Function;

@Configuration
public class TextProcessorCfg {

    /**
     * 文本处理函数 - 接收原始消息，处理后发送新消息
     * 这是一个Function函数，接收输入，产生输出
     */
    @Bean
    public Function<ProducerCfg.MessageEvent, ProcessedMessageEvent> textProcessor() {
        return input -> {
            System.out.println("🔄 Processing message: " + input.getMessageId());

            // 模拟处理逻辑
            String processedContent = processContent(input.getContent());
            int wordCount = countWords(input.getContent());
            String sentiment = analyzeSentiment(input.getContent());
            boolean containsKafka = input.getContent().toLowerCase().contains("kafka");

            ProcessedMessageEvent output = new ProcessedMessageEvent(
                    input.getMessageId(),
                    input.getContent(),
                    processedContent,
                    wordCount,
                    sentiment,
                    containsKafka,
                    LocalDateTime.now(),
                    "PROCESSED"
            );

            System.out.println("✅ Processed message: " + output.getMessageId() +
                    " | Words: " + output.getWordCount() +
                    " | Sentiment: " + output.getSentiment());

            return output;
        };
    }

    private String processContent(String content) {
        // 简单的处理逻辑：转大写，添加处理标记
        return "[PROCESSED] " + content.toUpperCase();
    }

    private int countWords(String content) {
        if (content == null || content.trim().isEmpty()) {
            return 0;
        }
        return content.split("\\s+").length;
    }

    private String analyzeSentiment(String content) {
        content = content.toLowerCase();
        if (content.contains("easy") || content.contains("good") || content.contains("great")) {
            return "POSITIVE";
        } else if (content.contains("hard") || content.contains("difficult") || content.contains("bad")) {
            return "NEGATIVE";
        } else {
            return "NEUTRAL";
        }
    }

    /**
     * 处理后的消息事件
     */
    public static class ProcessedMessageEvent {
        private String messageId;
        private String originalContent;
        private String processedContent;
        private int wordCount;
        private String sentiment;
        private boolean containsKafka;
        private LocalDateTime processedAt;
        private String status;

        // 构造器、getter、setter
        public ProcessedMessageEvent() {}

        public ProcessedMessageEvent(String messageId, String originalContent, String processedContent,
                                     int wordCount, String sentiment, boolean containsKafka,
                                     LocalDateTime processedAt, String status) {
            this.messageId = messageId;
            this.originalContent = originalContent;
            this.processedContent = processedContent;
            this.wordCount = wordCount;
            this.sentiment = sentiment;
            this.containsKafka = containsKafka;
            this.processedAt = processedAt;
            this.status = status;
        }

        // Getters and Setters
        public String getMessageId() { return messageId; }
        public void setMessageId(String messageId) { this.messageId = messageId; }

        public String getOriginalContent() { return originalContent; }
        public void setOriginalContent(String originalContent) { this.originalContent = originalContent; }

        public String getProcessedContent() { return processedContent; }
        public void setProcessedContent(String processedContent) { this.processedContent = processedContent; }

        public int getWordCount() { return wordCount; }
        public void setWordCount(int wordCount) { this.wordCount = wordCount; }

        public String getSentiment() { return sentiment; }
        public void setSentiment(String sentiment) { this.sentiment = sentiment; }

        public boolean isContainsKafka() { return containsKafka; }
        public void setContainsKafka(boolean containsKafka) { this.containsKafka = containsKafka; }

        public LocalDateTime getProcessedAt() { return processedAt; }
        public void setProcessedAt(LocalDateTime processedAt) { this.processedAt = processedAt; }

        public String getStatus() { return status; }
        public void setStatus(String status) { this.status = status; }

        @Override
        public String toString() {
            return "ProcessedMessageEvent{" +
                    "messageId='" + messageId + '\'' +
                    ", originalContent='" + originalContent + '\'' +
                    ", processedContent='" + processedContent + '\'' +
                    ", wordCount=" + wordCount +
                    ", sentiment='" + sentiment + '\'' +
                    ", containsKafka=" + containsKafka +
                    ", processedAt=" + processedAt +
                    ", status='" + status + '\'' +
                    '}';
        }
    }
}
